package tman.system.peer.tman;

import common.configuration.CyclonConfiguration;
import common.configuration.TManConfiguration;
import common.peer.DescriptorBuffer;
import common.peer.PeerAddress;
import common.peer.PeerDescriptor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;

import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.network.Network;
import se.sics.kompics.timer.CancelTimeout;
import se.sics.kompics.timer.SchedulePeriodicTimeout;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;
import common.peer.PeerData;
import common.configuration.Global;
import common.events.PeerDataRequest;
import common.events.PeerDataResponse;
import common.simulation.PeerStatus;
import cyclon.system.peer.cyclon.CyclonPartnersPort;
import cyclon.system.peer.cyclon.CyclonPartnersRequest;
import cyclon.system.peer.cyclon.CyclonPartnersResponse;

public final class TMan extends ComponentDefinition {

    Negative<PeerStatus> peerStatusPort = negative(PeerStatus.class);
    Positive<CyclonPartnersPort> cyclonPartnersPort = positive(CyclonPartnersPort.class);
    Positive<Network> networkPort = positive(Network.class);
    Positive<Timer> timerPort = positive(Timer.class);
	//private PeerDescriptor self;
	private HashMap<String, Integer> eventCounts;
	private long shuffleTimeout;
	private Cache cache;
	private boolean joining;
	long period;
	private TManConfiguration TManConfiguration;
	private HashMap<UUID, PeerAddress> outstandingRandomShuffles;

//-------------------------------------------------------------------	
	public TMan() {
		outstandingRandomShuffles = new HashMap<UUID, PeerAddress>();
		eventCounts = new HashMap<String, Integer>();

        subscribe(handleInit, control);
        subscribe(handleRound, timerPort);
        subscribe(handleBufferTimeout, timerPort);
        subscribe(handleCyclonPartnersResponse, cyclonPartnersPort);
        subscribe(handleTManBufferRequest, networkPort);
        subscribe(handleTManBufferResponse, networkPort);
        subscribe(handleAcquireParent, networkPort);
        subscribe(handleParentResponse, networkPort);
        subscribe(handleUpdateCost, networkPort);
        subscribe(handleOrphanSelf, networkPort);
        subscribe(handleOrphanChild, networkPort);
        subscribe(handlePeerDataRequest, peerStatusPort);
	}

//-------------------------------------------------------------------	
	Handler<TManInit> handleInit = new Handler<TManInit>() {
		public void handle(TManInit init) {
			joining = true;
			//self = new PeerDescriptor(init.getSelf());
			cache = new Cache(Global.tvs, new PeerDescriptor(init.getSelf()));
            TManConfiguration = init.getConfiguration();
            period = TManConfiguration.getPeriod();
            SchedulePeriodicTimeout rst = new SchedulePeriodicTimeout(period, period);
            rst.setTimeoutEvent(new TManSchedule(rst));
            trigger(rst, timerPort);

    }
};

//-------------------------------------------------------------------  
Handler<TManSchedule> handleRound = new Handler<TManSchedule>() {
    public void handle(TManSchedule event) {
            CyclonPartnersRequest response = new CyclonPartnersRequest();
            if(joining)
            	response.setSelf(cache.getSelf());
            trigger(response, cyclonPartnersPort);
    }
};

//-------------------------------------------------------------------Edited
Handler<PeerDataRequest> handlePeerDataRequest = new Handler<PeerDataRequest>() {
	public void handle(PeerDataRequest event) {
		PeerData pd = new PeerData(cache.getSelf().getPeerAddress(), toPeerAddressList(cache.getParents()), toPeerAddressList(cache.getChildren()), toPeerAddressList(cache.getAllCyclonPartners()), toPeerAddressList(cache.getAllSimilarEntries()), cache.getAverageCyclonAge(), cache.getAverageSimilarAge(), cache.getSelf().getCost());
		pd.setEventCounts(eventCounts);
		pd.setSp(cache.getSelf().getSP());

		for(String eventType: eventCounts.keySet())
		{
			eventCounts.put(eventType, 0);
		}
		PeerDataResponse responseEvent = new PeerDataResponse(event, pd);
		trigger(responseEvent, peerStatusPort);
	}
};

//-------------------------------------------------------------------  
Handler<CyclonPartnersResponse> handleCyclonPartnersResponse = new Handler<CyclonPartnersResponse>() {
        public void handle(CyclonPartnersResponse event) {
        		//Increment Age of Similar View entries
        		cache.incrementDescriptorAges();  
    			//System.out.println(event.getPartners());
        		//Merge Cyclon View       		
        		cache.bootstrapTManView(event.getPartners(), joining);
        		if(joining == true)
        			joining = false;
        		        		     		
        		//Schedule TMan Buffer Request
    			PeerAddress randomPeer = cache.selectPeerToShuffleWith();
    			//But only if there's at least one peer available
    			//randomPeer = null;
    			if(randomPeer != null)
    			{
	        		ArrayList<PeerDescriptor> randomDescriptors = cache.selectToSendAtActive(Global.tsl-1);
	        		randomDescriptors.add(cache.getSelf().copyPD());
	        		DescriptorBuffer randomBuffer = new DescriptorBuffer(cache.getSelf().getPeerAddress(), randomDescriptors);
	        		//Timeout if peer doesn't respond back
	        		ScheduleTimeout rst = new ScheduleTimeout(shuffleTimeout);
	        		rst.setTimeoutEvent(new TManBufferRequestTimeout(rst, randomPeer));
	        		UUID rTimeoutId = rst.getTimeoutEvent().getTimeoutId();
	        		outstandingRandomShuffles.put(rTimeoutId, randomPeer);
	        		TManBufferRequest rRequest = new TManBufferRequest(rTimeoutId, randomBuffer, cache.getSelf().getPeerAddress(), randomPeer);
	        		trigger(rst, timerPort);
	        		trigger(rRequest, networkPort);
    			}
        		

        }
};

//-------------------------------------------------------------------	
	Handler<TManBufferRequest> handleTManBufferRequest = new Handler<TManBufferRequest>() {
		public void handle(TManBufferRequest event) {
			//Do nothing if I'm still in the process of joining (i.e. my similar view is still being boostrapped)
			if(!joining)
			{
				PeerAddress peer = event.getPeerSource();
				DescriptorBuffer receivedTManBuffer = event.getRandomBuffer();
				ArrayList<PeerDescriptor> toSendTManDescriptors = cache.selectToSendAtPassive(Global.tsl-1);
				toSendTManDescriptors.add(cache.getSelf().copyPD());
				DescriptorBuffer toSendTManBuffer = new DescriptorBuffer(cache.getSelf().getPeerAddress(), toSendTManDescriptors);
				//Acquire parents if necessary
				ArrayList<PeerDescriptor> peersToAcquire = cache.selectToKeep(peer, receivedTManBuffer.getDescriptors());
				for(PeerDescriptor pta: peersToAcquire)
				{
					countEvent("AcquireParent");
					TManAcquireParent request = new TManAcquireParent(cache.getSelf().getPeerAddress(), pta.getPeerAddress(), cache.getSelf().copyPD());
					trigger(request, networkPort);
				}
				TManBufferResponse response = new TManBufferResponse(event.getRequestId(), toSendTManBuffer, cache.getSelf().getPeerAddress(), peer);
				trigger(response, networkPort);
			}
		}
	};

//-------------------------------------------------------------------	
	Handler<TManBufferResponse> handleTManBufferResponse = new Handler<TManBufferResponse>() {
		public void handle(TManBufferResponse event) {

			// cancel shuffle timeout
			UUID shuffleId = event.getRequestId();
			if (outstandingRandomShuffles.containsKey(shuffleId)) {
				outstandingRandomShuffles.remove(shuffleId);
				CancelTimeout ct = new CancelTimeout(shuffleId);
				trigger(ct, timerPort);
			}

			PeerAddress peer = event.getPeerSource();
			DescriptorBuffer receivedRandomBuffer = event.getRandomBuffer();
			ArrayList<PeerDescriptor> peersToAcquire = cache.selectToKeep(peer, receivedRandomBuffer.getDescriptors());
			for(PeerDescriptor pta: peersToAcquire)
			{
				countEvent("AcquireParent");
				TManAcquireParent request = new TManAcquireParent(cache.getSelf().getPeerAddress(), pta.getPeerAddress(), cache.getSelf().copyPD());
				trigger(request, networkPort);
			}
		}
	};
	
//------------------------------------------------------------------
	Handler<TManAcquireParent> handleAcquireParent = new Handler<TManAcquireParent>() {
		public void handle(TManAcquireParent event) {
			boolean decision = cache.decideOnChild(event.getChild());
			if(decision)
			{
				TManParentResponse response = new TManParentResponse(cache.getSelf().getPeerAddress(), event.getChild().getPeerAddress(), cache.getSelf().copyPD());
				trigger(response, networkPort);
				// Check & Remove unwanted children
				PeerAddress childToBeRemoved = cache.getChildToBeRemoved();
				if(childToBeRemoved != null)
				{
					//if(cache.getSelf().getPeerAddress().isSource())
					//	System.out.println("Removing: " + childToBeRemoved);
					countEvent("OrphanChild");
					TManOrphanChild notify = new TManOrphanChild(cache.getSelf().getPeerAddress(), childToBeRemoved, cache.getSelf().copyPD());
					trigger(notify, networkPort);
				}
			}
		}
	};
	
//------------------------------------------------------------------
	Handler<TManParentResponse> handleParentResponse = new Handler<TManParentResponse>() {
		public void handle(TManParentResponse event) {
			boolean decision = cache.stillWantParent(event.getParent());		
			//Accept parent
			if(decision)
			{
				//Check & Remove unwanted parents
				PeerAddress parentToBeRemoved = cache.getParentToBeRemoved();
				if(parentToBeRemoved != null)
				{
					countEvent("OrphanSelf");
					TManOrphanSelf notify = new TManOrphanSelf(cache.getSelf().getPeerAddress(), parentToBeRemoved, cache.getSelf().copyPD());
					trigger(notify, networkPort);
				}
				if(cache.hasCostChanged())
				{
					//Send Cost Update to children (if any exist)
					for(PeerDescriptor child: cache.getChildren())
					{
						countEvent("UpdateCost");
						TManUpdateCost updateNotification = new TManUpdateCost(cache.getSelf().getPeerAddress(), child.getPeerAddress(), cache.getSelf().copyPD());
						trigger(updateNotification, networkPort);
					}
				}
			}
			//Reject parent
			else
			{
				countEvent("OrphanSelf");
				TManOrphanSelf notify = new TManOrphanSelf(cache.getSelf().getPeerAddress(), event.getParent().getPeerAddress(), cache.getSelf().copyPD());
				trigger(notify, networkPort);
			}
		}
	};
	
//-------------------------------------------------------------------
	Handler<TManOrphanChild> handleOrphanChild = new Handler<TManOrphanChild>() {
		public void handle (TManOrphanChild event) {
			cache.removeParent(event.getParent());
			if(cache.hasCostChanged())
			{
				//Send Cost Update to children (if any exist)
				for(PeerDescriptor child: cache.getChildren())
				{
					countEvent("UpdateCost");
					TManUpdateCost updateNotification = new TManUpdateCost(cache.getSelf().getPeerAddress(), child.getPeerAddress(), cache.getSelf().copyPD());
					trigger(updateNotification, networkPort);
				}
			}
		}
	};
//-------------------------------------------------------------------
	Handler<TManOrphanSelf> handleOrphanSelf = new Handler<TManOrphanSelf>() {
		public void handle (TManOrphanSelf event) {
			cache.removeChild(event.getChild());
		}
	};
	
//-------------------------------------------------------------------
	Handler<TManUpdateCost> handleUpdateCost = new Handler<TManUpdateCost>() {
		public void handle (TManUpdateCost event) {
			cache.updateParent(event.getParent());
			if(cache.hasCostChanged())
			{
				//Send Cost Update to children (if any exist)
				for(PeerDescriptor child: cache.getChildren())
				{
					countEvent("UpdateCost");
					TManUpdateCost updateNotification = new TManUpdateCost(cache.getSelf().getPeerAddress(), child.getPeerAddress(), cache.getSelf().copyPD());
					trigger(updateNotification, networkPort);
				}
			}
		}
	};
//-------------------------------------------------------------------	
	Handler<TManBufferRequestTimeout> handleBufferTimeout = new Handler<TManBufferRequestTimeout>() {
		public void handle(TManBufferRequestTimeout event) {
			//This handler should remove the sent-to entries for the target peer from the cache
		}
	};
	
//-------------------------------------------------------------------
	private ArrayList<PeerDescriptor> getPartners() {
		ArrayList<PeerDescriptor> partnersDescriptors = cache.getAllCyclonPartners();
		ArrayList<PeerDescriptor> partners = new ArrayList<PeerDescriptor>();
		for (PeerDescriptor desc : partnersDescriptors)
		{
			partners.add(desc);
		}
		
		return partners;
	}
	
	private void countEvent(String eventName)
	{
		if(eventCounts.get(eventName)==null)
			eventCounts.put(eventName, 1);
		else eventCounts.put(eventName, eventCounts.get(eventName)+1);
	}
	
	private ArrayList<PeerAddress> toPeerAddressList(ArrayList<PeerDescriptor> map)
	{
		ArrayList<PeerAddress> myList = new ArrayList<PeerAddress>();
		for (PeerDescriptor desc : map)
		{
			myList.add(desc.getPeerAddress());
		}
		
		return myList;	
	}
	
	private ArrayList<PeerAddress> fromMapToPeerAddressList(HashMap<Integer, PeerDescriptor> map)
	{
		ArrayList<PeerAddress> myList = new ArrayList<PeerAddress>();
		for (Integer BW : map.keySet())
		{
			myList.add(map.get(BW).getPeerAddress());
		}
		
		return myList;	
	}
}
